package org.teemor.demo.kafka.springclient.producer;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.concurrent.ExecutionException;

/**
 * @Description:
 * @author zhoulk
 * @date: 2020/2/12
 */
@Component
public class SimpleProducer {

    private static final Logger LOGGER = LoggerFactory.getLogger(SimpleProducer.class);

    @Autowired
    KafkaTemplate<Integer, String> kafkaTemplate;

    public void simpleProduce(String topic, String data) throws ExecutionException, InterruptedException {
        ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(topic, 123, data);
        future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                LOGGER.error("发送失败");
            }

            @Override
            public void onSuccess(SendResult<Integer, String> integerStringSendResult) {
                LOGGER.info("发送成功");
            }
        });
        SendResult<Integer, String> result = future.get();
        System.out.println(result.toString());
        LOGGER.info("方法执行结束");
    }

    public void defaultProduce(Integer partition, String data) {
        ListenableFuture<SendResult<Integer, String>> result = kafkaTemplate.sendDefault(partition, data);
        // ...
    }
    public void recordProduce(ProducerRecord<Integer, String> record) throws ExecutionException, InterruptedException {
        ListenableFuture<SendResult<Integer, String>> result = kafkaTemplate.send(record);
        result.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println("发送失败");
            }

            @Override
            public void onSuccess(SendResult<Integer, String> integerStringSendResult) {
                System.out.println("发送成功");
            }
        });
        SendResult<Integer, String> result1 = result.get();
        System.out.println(result1.toString());
    }
}
